package top.lethelh.jianoa.task;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import top.lethelh.jianoa.db.pojo.MessageEntity;
import top.lethelh.jianoa.db.pojo.MessageRefEntity;
import top.lethelh.jianoa.exception.JoaException;
import top.lethelh.jianoa.service.MessageService;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /**
     * 同步消息发送
     * @param topic 绑定的topic
     * @param entity 发送的消息实体
     */
    public void send(String topic, MessageEntity entity) {
        // 首先往MongoDB插入消息实体
        String id = messageService.insertMessage(entity);
        // 创建连接
        try (
                // 新建一个连接对象
                Connection con = factory.newConnection();
                // 给这个连接创建一个频道
                Channel channel = con.createChannel();
        ) {
            // 通过设置参数的方式，连接到主题
            channel.queueDeclare(topic, true, false, false, null);
            // 创建一个HashMap来设置消息信息
            HashMap<String, Object> msgMap = new HashMap<>();
            // 设置消息编号
            msgMap.put("messageId", id);
            // 创建一个AMQP，并给其赋一些属性
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(msgMap).build();
            // 以字符集UTF-8的方式发布消息到主题上
            channel.basicPublish("", topic, properties, entity.getMsg().getBytes(StandardCharsets.UTF_8));
            // 控制台打印一些标志
            log.info("Msg send success.");
        } catch (Exception e) {
            // 发布失败
            log.error("Error!", e);
            // 抛出业务异常
            throw new JoaException("Send to MQ error!");
        }
    }

    /**
     * 异步发送消息 使用@Async注解
     * @param topic 消息绑定的主题
     * @param msgEntity 发送的消息实体
     */
    @Async
    public void sendAsync(String topic, MessageEntity msgEntity) {
        send(topic, msgEntity);
    }

    /**
     * 从MQ中接收消息(Sync)
     * @param topic 从那个主题获取
     * @return 收到的消息数量
     */
    public int receive(String topic) {
        // 首先定义接收的消息数量，初始为0
        int msgNum = 0;
        // 创建MQ连接
        try (
                // 新建一个连接对象
                Connection con = factory.newConnection();
                // 给这个连接创建一个频道
                Channel channel = con.createChannel();
        ) {
            // 从传入的主题里获取消息，之后几个是设置接收的参数
            channel.queueDeclare(topic, true, false, false, null);
            // 使用while死循环来轮询主题
            while (true) {
                // 获得到MQ里的一条响应
                GetResponse resp = channel.basicGet(topic, false);
                // 如果响应对象不为空（也就是存在）
                if (resp != null) {
                    // 从响应里获得消息参数
                    AMQP.BasicProperties prop = resp.getProps();
                    // 以key-value的形式从headers里获取参数
                    Map<String, Object> res = prop.getHeaders();
                    // 从res里获得messageId
                    String messageId = res.get("messageId").toString();
                    // 从响应里获得消息主体
                    byte[] msgBody = resp.getBody();
                    // 将byte型的消息信息转换成字符串
                    String msg = new String(msgBody);
                    log.info("From MQ message is:" + msg);

                    // 创建一个消息引用对象来保存 消息已读未读状态
                    MessageRefEntity mrEntity = new MessageRefEntity();
                    // 设置messageId
                    mrEntity.setMessageId(messageId);
                    // 设置接收消息的用户id
                    mrEntity.setReceiverId(Integer.parseInt(topic));
                    // 设置为未读
                    mrEntity.setReadFlag(false);
                    // 设置最新消息
                    mrEntity.setLastFlag(true);
                    // 往MongoDB插入消息引用对象
                    messageService.insertRef(mrEntity);

                    // 当上述步骤完成之后，发送ack应答，提醒MQ可以删除这条消息了
                    long tag = resp.getEnvelope().getDeliveryTag();
                    channel.basicAck(tag, false);
                    // 接收的消息数量递增+1
                    msgNum++;
                } else {
                    // MQ里没有消息了，则跳出循环
                    break;
                }
            }
        } catch (Exception e) {
            // 遇到错误，则在控制台日志输出
            log.error("Error", e);
            // 抛出一个新的业务异常
            throw new JoaException("Received msg from MQ error!");
        }
        // 最终返回接收的消息数量
        return msgNum;
    }

    /**
     * 异步接收消息 使用@Async注解
     * @param topic 消息绑定的主题
     * @return 接收的消息数量
     */
    @Async
    public int receiveAsync(String topic) {
        return receive(topic);
    }

    public void deleteQueue(String topic) {
        try (
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();
        ) {
            channel.queueDelete(topic);
            log.debug("MQ队列成功删除");

        } catch (Exception e) {
            log.error("删除队列失败", e);
            throw new JoaException("删除队列失败");
        }
    }

    @Async
    public void deleteQueueAsync(String topic) {
        deleteQueue(topic);
    }
}
